#coding=utf-8
# https://blog.csdn.net/qhd1994/article/details/79864087

# **********
'''
多进程 Multiprocessing 模块
Process 类
Process 类用来描述一个进程对象。创建子进程的时候，只需要传入一个执行函数和函数的参数即可完成 Process 示例的创建。

p = multiprocessing.Process(group=None, target=None, name=None, args=(), kwargs={}, *, daemon=None)

方法：
p.start() 方法启动进程
p.join()  方法实现进程间的同步，等待所有进程退出。
p.terminate() 方法实现进程的强制终止

参数：
target 是函数名字，需要调用的函数
args 函数需要的参数，以 tuple 的形式传入
'''
import multiprocessing
import os
import time
import random

def run_proc(name):
    print ('child process {0} {1}'.format(name,os.getpid()))


def main_process_run_proc():
    print('Parent process {0} is Running'.format(os.getpid()))
    for i in range(5):
        p = multiprocessing.Process(target=run_proc,args=(str(i),))
        print 'process start'
        p.start()
    p.join()
    print 'process close'


# **********
'''
Pool
Pool 可以提供指定数量的进程供用户使用，默认是 CPU 核数。当有新的请求提交到 Poll 的时候，如果池子没有满，会创建一个进程来执行，否则就会让该请求等待。
- Pool 对象调用 join 方法会等待所有的子进程执行完毕
- 调用 join 方法之前，必须调用 close
- 调用 close 之后就不能继续添加新的 Process 了，用来阻止多余的进程涌入进程池 Pool 造成进程阻塞。
'''

# **********
# pool.apply_async
# apply_async 方法用来同步执行进程，允许多个进程同时进入池子。
def run_task_apply_async(name):
    print('Task {0} pid {1} is running, parent id is {2}'.format(name,os.getpid(),os.getppid()))
    time.sleep(1)
    print ('Task {0} end.'.format(name))

def main_pool_run_task_apply_async():
    print('current process {0}'.format(os.getpid()))
    pool = multiprocessing.Pool(processes=3)
    # 创建一个运行同时有3个进程运行的池子
    for i in range(6):
        print 'add ',i
        pool.apply_async(run_task_apply_async,args=(i,))
        # 具此print信息观察，用apply_async时，会一次性将6个进程都加入到pool内，
        # 只是pool因为规定，只允许同时执行3个，相当于另外三个是在pool的缓存区，而不自执行区
        # pool执行区内的进程的执行结束1个，就从缓存区调一个到执行区并立即开始执行
        # 直到所有的都结束
    print('waiting for all subprocess done ...')
    pool.close() # 关闭池子，不再有新的进程加入
    pool.join()  # 等待池子内所有进程运行结束
    print('all processes done!')


# **********
# pool.apply
# pool.apply(func[, args[, kwds]])
# 该方法只能允许一个进程进入池子，在一个进程结束之后，另外一个进程才可以进入池子。
def run_task_apply(name):
    print('Task {0} pid {1} is running, parent id is {2}'.format(name, os.getpid(), os.getppid()))
    time.sleep(1)
    print('Task {0} end.'.format(name))

def main_pool_run_task_apply():
    print('current process {0}'.format(os.getpid()))
    p = multiprocessing.Pool(processes=3)
    for i in range(6):
        print 'add ',i
        p.apply(run_task_apply, args=(i,))
        # 具此print信息观察，用apply时，会一次性只能将1个进都加入到pool内，并立即执行
        # 表现出阻塞效果，别的进程将无法进入到pool内
        # 从效果看，此时的pool相当于只有一个执行区，而且只能执行1个，没有缓存队列区
    print('Waiting for all subprocesses done...')
    p.close() # pool不在接收新的进程
    p.join()  # 等待pool内的进程都运行结束
    print('All processes done!')


# **********
# pool.map方法
# 上面我们是通过一个循环往进程池添加任务，Pool提供了更优雅的map方法来管理任务的提交。
def task_map(msg):
    print('Task {0} pid {1} is running, parent id is {2}'.format(msg, os.getpid(), os.getppid()))
    print 'hello, %s' % msg
    time.sleep(1)
    print('Task {0} end.'.format(msg))
    return 'msg: %s' % msg

def main_pool_task_map():
    pool = multiprocessing.Pool(processes=4)

    results = []
    msgs = [x for x in range(10)]
    results = pool.map(task_map, msgs)
    # 从效果看，其表现类似于pool.apply_async，只是一次性将task给与pool
    # pool.map会将msgs list的内容一个个的调入到pool的执行区内并执行，但是并不超过声明的pool的进程容量
    # 依次序列 顺序执行，知道所有的都键入
    pool.close() # pool关闭，不再接收新的进程进入
    pool.join()  # 等待pool整个队列都结束掉
    print 'processes done, result:'
    for x in results:
        print x


# **********
# Queue 进程间通信
# Queue 用来在多个进程间通信。Queue 有两个方法，get 和 put。

# **********
'''
put 方法
queue.put(self, obj, block=True, timeout=None)
    Put 方法用来插入数据到队列中，有两个可选参数，blocked 和 timeout。
    - blocked = True（默认值），timeout 为正
      该方法会阻塞 timeout 指定的时间，直到该队列有剩余空间。如果超时，抛出 Queue.Full 异常。
    - blocked = False
      如果 Queue 已满， 立刻抛出 Queue.Full 异常

get 方法
queue.get(self, block=True, timeout=None)
    get 方法用来从队列中读取并删除一个元素。有两个参数可选，blocked 和 timeout
    - blocked = False （默认），timeout 正值
      等待时间内，没有取到任何元素，会抛出 Queue.Empty 异常。
    - blocked = True
      Queue 有一个值可用，立刻返回改值；Queue 没有任何元素，阻塞等待
'''
# 写数据进程执行的代码:
def proc_write_queue_put(queue,urls):
    print('Process(%s) is writing...' % os.getpid())
    for url in urls:
        queue.put(url)
        print('Put {} to queue... process {}'.format(url,os.getpid()))
        time.sleep(random.random())
# 读数据进程执行的代码:
def proc_read_queue_get(queue):
    print('Process(%s) is reading...' % os.getpid())
    while True:
        url = queue.get()
        print('Get {1} from queue. process {0}'.format(os.getpid(),url))
def main_queue_task_proc():
    # 父进程创建Queue，并传给各个子进程
    q = multiprocessing.Queue()
    proc_writer1 = multiprocessing.Process(target=proc_write_queue_put, args=(q,['url_1', 'url_2', 'url_3']))
    proc_writer2 = multiprocessing.Process(target=proc_write_queue_put, args=(q,['url_4','url_5','url_6']))
    proc_reader = multiprocessing.Process(target=proc_read_queue_get, args=(q,))
    proc_reader1 = multiprocessing.Process(target=proc_read_queue_get, args=(q,))
    # 启动子进程proc_writer，写入:
    proc_writer1.start() # queue 在效果上==堆栈，此相当于push端
    proc_writer2.start() # queue 在效果上==堆栈，此相当于push端
    # 启动子进程proc_reader，读取:
    proc_reader.start() # queue 在效果上==堆栈，此相当于pop端
    # proc_reader1.start()
    # 等待proc_writer结束:
    proc_writer1.join()
    proc_writer2.join()
    # proc_reader进程里是死循环，无法等待其结束，只能强行终止
    proc_reader.terminate()
    # 队列，相当于一个FIFO的堆栈一样，对于向队列压入信息的进程个数没有限制
    # 同样，对于从队列out端弹出信息的进程个数也没有限制
    # 相当于1:N，或者N:1，或者N:N的形态
    # 对于N个进程的情况，无论压入端还是弹出端，消息的顺序并不确定，完全是由系统根据每个进程执行的空闲顺序而竞争获取

    # 如果某个进程是死循环的输入，那么就可以认为建立了一个持续的且稳固的调用关系
    # 及一个进程什么状态，另一个进程响应的做出什么动作
    # 当然不可以实现以下的情况，及一个进程状态发生变化 却 要很多out端进程同时变化
    # 通过queue是做不到这一点的，因为堆栈的信息是被pop出来的，即有进程获取到此消息后，堆栈中就不再由此消息了


# **********
# Pipe 进程间通信
# 常用来在两个进程间通信，两个进程分别位于管道的两端。
# multiprocessing.Pipe([duplex])
# subprocess发送，parentprocess接收
def pipe_send(pipe):
    pipe.send(['spam'] + [42, 'egg'])   # send 传输一个列表, 3个元素
    pipe.close()

def main_pipe_send_task():
    (con1,con2) = multiprocessing.Pipe()
    sender = multiprocessing.Process(target=pipe_send,args=(con1,))
    sender.start()
    # print 'waiting...'
    print ("con2 got: %s" % con2.recv()) # con2 got: ['spam', 42, 'egg']， 获取到的也是个列表
    # 函数pipe.recv()是阻塞式的
    # 比如将 pipe_send 中的 pipe.send comment掉，那么此函数将一直停留在 上边的这个 print语句
    # 即著进程被阻塞
    con2.close()

# ----------
# subprocess发送 + 接收，parentprocess接收 + 发送
def pipe_talk(pipe):
    pipe.send(dict(name='Bob', spam=42))            # 传输一个字典
    reply = pipe.recv()                             # 接收传输的数据
    print 'talker got:', reply
    # 函数pipe.recv()是阻塞式的

def main_pipe_talk_task():
    (parentEnd, childEnd) = multiprocessing.Pipe() # 创建两个 Pipe() 实例，也可以改成 conf1， conf2
    child = multiprocessing.Process(target=pipe_talk, args=(childEnd,)) # 创建一个 Process 进程，名称为 child
    child.start() # 启动进程
    print 'parent got:', parentEnd.recv() # parentEnd 是一个 Pip() 管道，可以接收 child Process 进程传输的数据
    # 函数pipe.recv()是阻塞式的
    # child进程中用的是childEnd，即pipe的child端
    # 以上可知，child发送的信息，只能parentEnd收到，child端是无法收到自己发送的信息的
    # time.sleep(2)
    parentEnd.send([x*2 for x in 'spam']) # parentEnd 是一个 Pip() 管道，可以使用 send 方法来传输数据
    child.join() # 传输的数据被 talk 函数内的 pip 管道接收，并赋值给 reply
    print('parent exit')
    # 由此可知，parentEnd发送的信息，只能dhildEnd收到，parentEnd端是无法收到自己发送的信息的


# **********
# Lock
# 当多个进程需要访问共享资源的时候，Lock可以用来避免访问的冲突
# 要访问共享文件，先获得锁的进程会阻塞后面的进程
def task1(lock, f):
    with lock:
        f = open(f, 'w+')
        f.write('hello ')
        time.sleep(1)
        f.close()

def task2(lock, f):
    lock.acquire()
    try:
        f = open(f, 'a+')
        time.sleep(1)
        f.write('world!')
    except Exception as e:
        print(e)
    finally:
        f.close()
        lock.release()

def main_lock_task():
    lock = multiprocessing.Lock()
    fn = './file.txt'

    start = time.time()
    p1 = multiprocessing.Process(target=task1, args=(lock, fn,))
    p2 = multiprocessing.Process(target=task2, args=(lock, fn,))
    # 不同进程用同一个锁，同一个锁才能做到让不同进程互斥的操作
    p1.start()
    p2.start()
    p1.join()
    p2.join()

    end = time.time()
    print 'time cost: %s seconds' % (end - start)
    # 由lock的作用可知，task1和task2在启动的时候也许有极低的几率执行书序发生变化
    # 但是，每个函数一旦开始直径，必定会到此函数执行结束后，另一个函数才能执行
    # 这就是此处lock的作用
    # 也因此，此处统计出来的时间为2s左右，而不是1s
    with open(fn, 'r') as f:
        for x in f.readlines():
            print x


# **********
'''
Semaphore
Semaphore 和 Lock 稍有不同，Semaphore 相当于 N 把锁，获取其中一把就可以执行了。
信号量的总数 N 在构造时传入，s = Semaphore(N)。
和 Lock 一样，如果信号量为0，则进程堵塞，直到信号大于0。
Semaphore可用来控制对共享资源的访问数量，例如池的最大连接数。

信号量同步基于内部计数器，
每调用一次acquire()，计数器减1；
每调用一次release()，计数器加1.
当计数器为0时，acquire()调用被阻塞。
这是Dijkstra信号量概念P()和V()的Python实现。
信号量同步机制适用于访问像服务器、文件这样的有限资源。
'''
def task_semaphore(s, msg):
    s.acquire()
    print 'hello, %s' % msg
    time.sleep(1)
    s.release()

def main_semaphore():
    s=multiprocessing.Semaphore(2)
    processes=[]
    for x in range(8):
        p = multiprocessing.Process(target=task_semaphore,args=(s,x,))
        processes.append(p)
    start = time.time()
    for p in processes:
        p.start()
    for p in processes:
        p.join()
    end = time.time()
    print '8 process takes %s seconds' % (end - start)
    # 这样实现的效果，跟pool = multiprocessing.Pool(processes=3)是一致的
    # 每次只允许2个进程同时运行，结束一个启动一个，直到所有的进程都运行完毕


# **********
# Event
# Event用来实现进程间同步通信
# event.is_set()            ：返回event的状态值；
# event.wait(timeout=None) ：如果 event.isSet()==False将阻塞线程；
#                            当在线程中调用wait时，如果事件内部标识为False，则会阻塞，直到set方法被调用，将内部标识设置为True
# event.set()              ：设置event的状态值为True，所有阻塞池的线程激活进入就绪状态， 等待操作系统调度；
#                            Event对象最初创建时，内部标识默认是False
# event.clear()            ：将内部标识重新设置为初始默认值False, 所有的e.wait()重新阻塞起来
#
# 一个典型例子：交通红绿灯
# 假设路上每辆汽车都是一个单独的进程，而红绿灯就是那个event
# event.wait()，相当于汽车在路口观察下红绿灯，如果为红色，就停下来，如果未绿色就继续
# 如果event.set()后，即红绿灯变为绿灯后，所有的汽车都可以开动运行了
# 如果event.clear()后，即红绿灯变温红灯后，所有下一批次来的汽车都要重新被停下来
# event的状态值即为红绿灯，true为绿灯，false为红灯
def task1_event(e,timeout):
    print 'task1 is waitting...'
    e.wait() # 如果e.is_set()==Flase，则阻塞，否则继续向下执行
    time.sleep(1)
    print 'hello, timeout=%s, e.is_set(): %s' % (timeout, e.is_set())

def task2_event(e, timeout):
    print 'task2 is waitting...'
    e.wait(timeout) # msg为timeout 时间，单位为s
    print 'hello, timeout=%s, e.is_set(): %s' % (timeout, e.is_set())

def main_event():
    e = multiprocessing.Event()
    p1 = multiprocessing.Process(target=task1_event, args=(e, 1))
    p2 = multiprocessing.Process(target=task2_event, args=(e, 2))
    # 不同的process共用一个event对象，才能根据event的状态实现互斥操作
    p1.start()
    p2.start()
    time.sleep(3)
    e.set() # 只有执行了set()以后，其它进程中的e.wait()才能解除阻塞，并向下执行
    print 'main: event is set'
    # 由其原理可知，如果有>2个的进程同时处于e.wait()的状态，那么在他们都是处于争抢的状态
    # 即下一步谁会先运行，谁会后运行顺序不定
    #
    # 只能确保众多个process是一个个去访问共享资源，而不能确定谁先访问谁后访问


if __name__ == '__main__':
    # 单个进程的声明和启动
    main_process_run_proc()          # p = multiprocessing.Process()

    # 进程池的几种管理进程的方式，pool.apply_async，pool.apply，pool.map
    main_pool_run_task_apply_async() # multiprocessing.Pool.apply_async()
    main_pool_run_task_apply()       # multiprocessing.Pool.apply()
    main_pool_task_map()             # multiprocessing.Pool.map()

    # 进程间通信的2中方式：queue和pipe，
    # pipe返回2个对象，像是一个两端都可进可出的堆栈，提供了send和recv的方法访问内容
    # queue返回1个对象，像是将堆栈封装到了内部，提供了put和get方法访问堆栈
    main_queue_task_proc()           # q=multiprocessing.Queue(), q.put(), q.get()
    main_pipe_send_task()            # pin,pout = multiprocessing.Pipe()
    main_pipe_talk_task()

    # 这些lock，event和semaphore, 从表现上说都是单例对象，通过方法检查单例中变量或者计数器的状态
    # 在状态不符合时，使用while循环等待阻塞，今儿达到访问共享资源的可控

    main_lock_task()                 # multiprocessing.Lock()
    main_semaphore()                 # s=multiprocessing.Semaphore(2)
    main_event()                     # e = multiprocessing.Event()
